package com.zoe.flume.sink.elasticsearch;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.FastDateFormat;
import org.apache.flume.Context;
import org.apache.flume.conf.ComponentConfiguration;
import org.elasticsearch.common.Strings;
import org.joda.time.DateTimeUtils;

import java.util.TimeZone;

/**
 * Created by shenyuhang on 2018/2/1.
 */
public class DateTimeIndexNameBuilder {

  public static final String DATE_FORMAT = "dateFormat";
  public static final String TIME_ZONE = "timeZone";

  public static final String DEFAULT_DATE_FORMAT = "yyyy-MM-dd";
  public static final String DEFAULT_TIME_ZONE = "Etc/UTC";

  private FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd",
      TimeZone.getTimeZone("Etc/UTC"));

  private String indexPrefix;

  public String getIndexName() {
    long timestamp =  DateTimeUtils.currentTimeMillis();
    return new StringBuilder(indexPrefix).append('-')
      .append(fastDateFormat.format(timestamp)).toString();
  }

  public void configure(Context context) {
    String dateFormatString = context.getString(DATE_FORMAT);
    String timeZoneString = context.getString(TIME_ZONE);
    if (StringUtils.isBlank(dateFormatString)) {
      dateFormatString = DEFAULT_DATE_FORMAT;
    }
    if (StringUtils.isBlank(timeZoneString)) {
      timeZoneString = DEFAULT_TIME_ZONE;
    }
    fastDateFormat = FastDateFormat.getInstance(dateFormatString,
        TimeZone.getTimeZone(timeZoneString));
    indexPrefix = context.getString(ElasticSearchSinkConstants.INDEX_NAME);
    if (Strings.isNullOrEmpty(indexPrefix)) indexPrefix =  ElasticSearchSinkConstants.DEFAULT_INDEX_NAME;
  }

  public void configure(ComponentConfiguration conf) {
  }
}
